package com.kingjw.sourceTest

import java.util.{Properties, Random}

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011

/**
 *
 * @Package: com.kingjw.sourceTest
 * @ClassName: KafkaSource
 * @Author: 王坚伟
 * @CreateTime: 2022/1/17 15:35
 * @Description:
 */
object KafkaSource {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
//    val properties = new Properties()
//    properties.setProperty("bootstrap.servers", "hadoop117:9092")//重要
//    properties.setProperty("group.id", "consumer-group")//重要
//    properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
//    properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
//    properties.setProperty("auto.offset.reset", "latest")
//    val inputDS = env.addSource(new FlinkKafkaConsumer011[String]("sensor",new SimpleStringSchema(),properties))

    val inputDS: DataStream[SensorReading] = env.addSource(new MySensorSource)
//    val value = inputDS.flatMap(_.split(" ")).map((_, 1)).keyBy(0).sum(1)
    val value = inputDS.map((_, 1)).keyBy(0).sum(1)
    value.print("kafka")
    env.execute("kafka source")
  }

}

//case class SensorReading(id: String, timestamp: Long, temperature: Double)

class MySensorSource extends SourceFunction[SensorReading] {

  var running:Boolean = true
  override def run(sourceContext: SourceFunction.SourceContext[SensorReading]): Unit = {
    // 初始化一个随机数发生器
    val rand = new Random()
    var curTemp = 1.to(10).map(
      i => ("sensor_" + i, 65 + rand.nextGaussian() * 20)
    )
    while (running) {
      // 更新温度值
      curTemp = curTemp.map(
        //rand.nextGaussian()为高斯随机数，即正态分布
        t => (t._1, t._2 + rand.nextGaussian())
      )
      // 获取当前时间戳
      val curTime = System.currentTimeMillis()
      curTemp.foreach(
        //发出数据
        t => sourceContext.collect(SensorReading(t._1, curTime, t._2))
      )
      Thread.sleep(100)
    }
  }

  override def cancel(): Unit = {
    running = false

  }
}
